package cn.itcast.streaming.function.window;

import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.utils.DateUtil;
import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.LinkedList;

//对驾驶行程采样数据自定义窗口函数操作(某辆车15min内没有上报数据就划分窗口，窗口内的数据再按5min维度进行数据采集和分析）
public class TripDriveSimpleWindowFunction implements WindowFunction<ItcastDataObj,String[],String, TimeWindow> {
    @Override
    public void apply(String key, TimeWindow timeWindow, Iterable<ItcastDataObj> iterable, Collector<String[]> collector) throws Exception {
        System.out.println("窗口范围："+DateUtil.timestampToDataStr(timeWindow.getStart())+"~"+DateUtil.timestampToDataStr(timeWindow.getEnd()));
        //1、获取当前车辆窗口内所以数据
        LinkedList<ItcastDataObj> itcastDataObjsList = Lists.newLinkedList(iterable);
        //2、对窗口内数据进行排序
        itcastDataObjsList.sort(((o1, o2) -> {
            //如果第一个元素对象时间大于第二个元素对象时间，则倒叙排序(从小到大的顺序）
            if(o1.getTerminalTimeStamp()>o2.getTerminalTimeStamp()){
                return 1;
            }else if(o1.getTerminalTimeStamp()<o2.getTerminalTimeStamp()){
                return -1;
            }else {
                return 0;
            }
        }));
        //3、对窗口内的数据进行前后两辆比较，如果时间间隔没有超过5min,则将数据追加到一个集合中（拼接起来）
        ItcastDataObj firstObj = itcastDataObjsList.getFirst();
        ItcastDataObj lastObj = itcastDataObjsList.getLast();
        Long startTimestamp = firstObj.getTerminalTimeStamp(); //第一条是数据时间
        System.out.println("第一条数据的时间："+ DateUtil.timestampToDataStr(startTimestamp));
        Long endTimestamp = lastObj.getTerminalTimeStamp(); //最后一条数据时间
        System.out.println("最后一条数据的时间："+ DateUtil.timestampToDataStr(endTimestamp));

        //循环遍历窗口内的所有数据，且每隔五秒钟进行一次数据的采样
        //获取到每条数据的soc、mileage（累计里程）、speed（车速）、gps（经度+维度）、
        // terminalTime（终端时间）字段，然后将各自字段各自拼接到一个字符串变量中
        //soc:剩余电量百分比
        StringBuffer singleSoc = new StringBuffer(firstObj.getSoc());
        //mileage:总里程数
        StringBuffer singleMileage = new StringBuffer(String.valueOf(firstObj.getTotalOdometer()));
        //speed:速度
        StringBuffer singleSpeed = new StringBuffer(String.valueOf(firstObj.getVehicleSpeed()));
        //gps:地理位置
        StringBuffer singleGps = new StringBuffer(firstObj.getLng()+"|"+firstObj.getLat());
        //terminalTime：终端时间
        StringBuffer singleTerminalTime = new StringBuffer(firstObj.getTerminalTime());
        for (ItcastDataObj obj : itcastDataObjsList) {
            Long currentTimestamp = obj.getTerminalTimeStamp();
            if((currentTimestamp-startTimestamp) >= 5*1000*60 || currentTimestamp == endTimestamp){
                singleSoc.append(","+obj.getSoc());
                singleMileage.append(","+obj.getTotalOdometer());
                singleSpeed.append(","+obj.getVehicleSpeed());
                singleGps.append(","+obj.getLng() + "|" + obj.getLat());
                singleTerminalTime.append(","+obj.getTerminalTime());
            }
            startTimestamp = currentTimestamp;
        }
        //4、将拼接好的字符串返回
        String[] result = new String[7];
        result[0] = firstObj.getVin();
        //使用窗口内的第一条数据时间作为处理时间
        result[1] = String.valueOf(firstObj.getTerminalTimeStamp());//终端时间
        result[2] = String.valueOf(singleSoc);
        result[3] = String.valueOf(singleMileage);
        result[4] = String.valueOf(singleSpeed);
        result[5] = String.valueOf(singleGps);
        result[6] = String.valueOf(singleTerminalTime);
        collector.collect(result);
    }
}
